package com.lhd.app.sink;

import com.lhd.app.bean.PetTypeLabel;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.util.Map;
import java.util.stream.Collectors;

// 本地文件输出Sink
public class LocalFileSink extends RichSinkFunction<PetTypeLabel> {
    private transient BufferedWriter writer;
    private String outputPath;
    
    public LocalFileSink(String outputPath) {
        this.outputPath = outputPath;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 创建data目录
        File dataDir = new File("data");
        if (!dataDir.exists()) {
            dataDir.mkdirs();
        }
        // 创建输出文件
        File outputFile = new File(outputPath);
        writer = new BufferedWriter(new FileWriter(outputFile, true)); // 追加模式
    }
    
    @Override
    public void invoke(PetTypeLabel label, SinkFunction.Context context) throws Exception {
        String record = String.format("%s,%s,%s,%d%n",
            label.getUserId(),
            String.join("|", label.getPetLabels()),
            formatScores(label.getPetScores()),
            label.getUpdateTime());
        
        writer.write(record);
        writer.flush();
        
        System.out.println("Written to local file: " + record.trim());
    }
    
    @Override
    public void close() throws Exception {
        if (writer != null) {
            writer.close();
        }
        super.close();
    }
    
    private String formatScores(Map<String, Double> scores) {
        return scores.entrySet().stream()
            .map(entry -> entry.getKey() + ":" + String.format("%.2f", entry.getValue()))
            .collect(Collectors.joining("|"));
    }
}